package function.stream;

import java.util.concurrent.ForkJoinPool;
import java.util.stream.IntStream;

/**
 * 并行流
 *
 * @author 旭旭
 * @create 2018-08-18 17:46
 **/
public class ParallelStream {

    public static void main(String[] args) {
//        createParallelStream();
//        feature2();
//        feature3();
        feature4();

    }

    /**
     * 特性一 并行流线程数
     * 并行流线程数默认为cpu个数
     * 默认线程池
     */
    public static void createParallelStream(){
        IntStream.range(1, 100).parallel().forEach(ParallelStream::printDebug);
    }

    /**
     * 特性二 并行再串行 以最后一个流为准
     */
    private static void feature2(){
        IntStream.range(1, 100).parallel().peek(ParallelStream::printDebug).sequential().peek(ParallelStream::printDebug2).count();
    }

    /**
     * 特性三 默认线程池与设置默认线程数
     */
    private static void feature3(){
        System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","3");
        IntStream.range(1, 100).parallel().forEach(ParallelStream::printDebug);
    }

    /**
     * 特性四 自定义线程池 防止线程被阻塞
     */
    private static void feature4(){
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        forkJoinPool.submit(() -> IntStream.range(1, 100).parallel().forEach(ParallelStream::printDebug));
        forkJoinPool.shutdown();

        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    private static void printDebug(int i){
//        System.out.println(i);
        System.out.println(Thread.currentThread().getName() + "debug:" + i);
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static void printDebug2(int i){
        System.err.println(i);
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
